package com.hdaccp.hadoop.mr1.distinct;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 *去除重复
 * 
 */
public class DistinctMusic {

	public static class DistinctMapper extends Mapper<LongWritable,Text,Text,NullWritable>{

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			context.write(value,  NullWritable.get());
		}
		
	}

	public static class DistinctReducer extends Reducer<Text,NullWritable,Text,NullWritable>{

		@Override
		protected void reduce(Text key, Iterable<NullWritable> values,	Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			context.write(key, NullWritable.get());
		}
		
	}
	
	private static class TextDescComparator extends Text.Comparator{

		@Override
		public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
			// TODO Auto-generated method stub
			return -super.compare(b1, s1, l1, b2, s2, l2);
		}

		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			// TODO Auto-generated method stub
			return -super.compare(a, b);
		}	
		
	}
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// TODO Auto-generated method stub
		Configuration conf=new Configuration();
		FileSystem.get(conf).delete(new Path("/output2/music/file1"),true);
		Job job=Job.getInstance(conf, "distinct music");
		
		job.setNumReduceTasks(1);
		job.setJarByClass(DistinctMusic.class);
		job.setMapperClass(DistinctMapper.class);
		job.setReducerClass(DistinctReducer.class);//必须设置，否则不能去重
		
		//更换为降序比较
		job.setSortComparatorClass(TextDescComparator.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		FileInputFormat.addInputPath(job, new Path("/input2/file1"));
		FileOutputFormat.setOutputPath(job, new Path("/output2/music/file1"));
		System.exit(job.waitForCompletion(true)?0:1);
		
	}

}
